package demo.hadoop.tsdb;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.hbase.async.HBaseClient;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import net.opentsdb.core.Aggregator;
import net.opentsdb.core.Aggregators;
import net.opentsdb.core.DataPoint;
import net.opentsdb.core.DataPoints;
import net.opentsdb.core.Query;
import net.opentsdb.core.SeekableView;
import net.opentsdb.core.TSDB;
import net.opentsdb.meta.Annotation;
import net.opentsdb.utils.Config;

/**
 * OpenTSDB is a distributed, scalable Time Series Database (TSDB) written on
 * top of HBase. OpenTSDB was written to address a common need: store, index and
 * serve metrics collected from computer systems (network gear, operating
 * systems, applications) at a large scale, and make this data easily accessible
 * and graphable.
 * 
 * {@link http://opentsdb.net/docs/build/html/user_guide/tuning.html}
 * 
 * @author junying.han
 *
 */
public class OpenTsdbDemo {

    private static Logger logger = LoggerFactory.getLogger(OpenTsdbDemo.class);

    public static String zkQuorum = "192.168.60.12:2181,192.168.60.25:2181,192.168.60.26:2181,192.168.60.37:2181,192.168.60.38:2181,192.168.60.39:2181,192.168.60.40:2181";
    public static String cfgPath = "D:\\Stan\\workspace\\javaDemo\\Demo\\testHadoop\\src\\main\\java\\demo\\hadoop\\hbase\\tsdb\\open_tsdb.properties";
    public static HBaseClient client = new HBaseClient(zkQuorum);

    public TSDB tsdb;

    public static TSDB buildTsdb() throws IOException {
        Config config = new Config(cfgPath);
        return new TSDB(config);
    }

    public OpenTsdbDemo() {
        try {
            this.tsdb = buildTsdb();
        } catch (IOException e) {
            logger.error("init TSDB error, program exist", e);
            throw new RuntimeException("init TSDB error, program exist", e);
        }
    }

    @Test
    public void testAddPoint() {
        new OpenTsdbDemo();
        Map<String, String> tags = new HashMap<String, String>();
        tags.put(MetricTags.EXCEPTION_COUNT.api.getStr(), "/api/test/aa");
        tags.put(MetricTags.EXCEPTION_COUNT.app.getStr(), "anchor-sell-backend");
        tags.put(MetricTags.EXCEPTION_COUNT.host.getStr(), "192.168.3.67");
        tags.put(MetricTags.EXCEPTION_COUNT.level.getStr(), "ERROR");
        tags.put(MetricTags.EXCEPTION_COUNT.log_type.getStr(),
                "com.ximalaya.mainstay.rpc.proxy.RpcInvocationHandler:handleInvocation:71");
        Map<String, String> encodeTags = encodeTagValue(tags);
        tsdb.addPoint(MetricTags.EXCEPTION_COUNT.metric.getStr(), System.currentTimeMillis() / 1000, 2L, encodeTags);
//        Deferred<Object> deferred = tsdb.addPoint(MetricTags.EXCEPTION_COUNT.metric.getStr(),
//                System.currentTimeMillis() / 1000, 2L, encodeTags);
//        logger.info("Deferred Result:{}", deferred);
//        logger.info("Deferred Result toJSON:{}", JSONObject.toJSONString(deferred));
    }

    public static void main(String[] args) {
        OpenTsdbDemo demo = new OpenTsdbDemo();
        demo.testAddPoint();
    }

    @Test
    public void monitor_redis_insert() {
        new OpenTsdbDemo();
        String metric = "monitor_redis";
        final Map<String, String> tags = new HashMap<String, String>();
        tags.put(MetricTags.EXCEPTION_COUNT.app.getStr(), "testweb");
        tags.put(MetricTags.EXCEPTION_COUNT.host.getStr(), "192.168.3.67");
        tags.put("result", "success");
        tags.put("redisOper", "get111");
        tags.put("api", "/albums/get_batch");
        tags.put("redisIp", TSDBEncodeUtil.encode("192.168.1.174:6379"));
        tsdb.addPoint(metric, new Date().getTime(), 20, tags);
        try {
            Thread.sleep(2000l);// 让tsdb写进去
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("write complete...");

//        long timestamp = System.currentTimeMillis() - 30000000l;
//        DataPoints[] dataPoints = tsdb.find(timestamp, System.currentTimeMillis(), metric, tags,
//                Aggregators.MAX, false);
//        for (final DataPoints dataPoint : dataPoints) {
//            for (int i = 0; i < dataPoint.size(); i++) {
//                System.out
//                        .println("****************************** result is *****************************************");
//                System.out.println(dataPoint.longValue(i));
//            }
//        }
    }

    @Test
    public void testFind() {
        OpenTsdbDemo demo = new OpenTsdbDemo();
        long now = System.currentTimeMillis();
        Map<String, String> tags = new HashMap<String, String>();
        tags.put("api", "*");
//        tags.put(MetricTags.EXCEPTION_COUNT.app.getStr(), "*");
//        tags.put(MetricTags.EXCEPTION_COUNT.host.getStr(), "*");
//        tags.put(MetricTags.EXCEPTION_COUNT.level.getStr(), "*");
//        tags.put(MetricTags.EXCEPTION_COUNT.log_type.getStr(), "*");
        logger.info("tags:{}", tags);
        Map<String, String> encodeTags = encodeTagValue(tags);
        logger.info("encodeTags:{}", encodeTags);
        DataPoints[] dataPointsArray = demo.query(now - 1000 * 3600 * 24 * 100, now, "exception_count", encodeTags,
                Aggregators.SUM, false);
        for (DataPoints dataPoints : dataPointsArray) {
            printDataPoints(dataPoints);
        }
    }

    public static Map<String, String> encodeTagValue(Map<String, String> tags) {
        Map<String, String> map = new HashMap<String, String>();
        if (tags == null || tags.isEmpty()) {
            return map;
        }
        for (Entry<String, String> tag : tags.entrySet()) {
            map.put(tag.getKey(), TSDBEncodeUtil.encode(tag.getValue()));
        }
        return map;
    }

    public void printDataPoints(DataPoints dataPoints) {
        logger.info("----------------------------DataPoints----------------------------");
        logger.info("dataPoints: {}", dataPoints);
        try {
            Map<String, String> tags = dataPoints.getTags();
            logger.info("tags:{}", tags);
        } catch (Exception e) {
            logger.error("", e);
        }
        try {
            List<Annotation> annotation = dataPoints.getAnnotations();
            logger.info("annotation:{}", annotation);
        } catch (Exception e) {
            logger.error("", e);
        }
        try {
            List<String> aggregatedTags = dataPoints.getAggregatedTags();
            logger.info("aggregatedTags : {}", aggregatedTags);
        } catch (Exception e) {
            logger.error("", e);
        }
        SeekableView seekableView = dataPoints.iterator();
        while (seekableView.hasNext()) {
            DataPoint dataPoint = seekableView.next();
            printDataPoint(dataPoint);
        }
    }

    public static void printDataPoint(DataPoint dataPoint) {
        logger.info("dataPoint : {}", dataPoint);
        if (dataPoint == null) {
            return;
        }
        long timestamp = dataPoint.timestamp();
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
        String time = format.format(new Date(timestamp));
        if (dataPoint.isInteger()) {
            long value = dataPoint.longValue();
            logger.info("timestamp:{},time:{},value:{}", timestamp, time, value);
        } else {
            double value = dataPoint.doubleValue();
            logger.info("timestamp:{},time:{},value:{}", timestamp, time, value);
        }
    }

    /**
     * 
     * @param start
     * @param end
     * @param metric
     * @param interval
     * @param downsampler 可选的时间间隔和函数，用于减少随时间返回的数据点的数量
     * @param tags
     * @param function
     * @param rate
     * @return
     */
    public DataPoints[] find(long start, long end, String metric, long interval, Aggregator downsampler,
            Map<String, String> tags, Aggregator function, boolean rate) {
        Query query = tsdb.newQuery();
        query.setStartTime(start);
        query.setEndTime(end);
        query.downsample(interval, downsampler);
        query.setTimeSeries(metric, tags, function, rate);
        return query.run();
    }

    /**
     * 
     * @param start    查询的开始时间。可以是绝对时间或相对时间
     * @param end      查询的结束时间。如果未提供结束时间，则当前时间即结束时间
     * @param metric   系统中的metric全名。必须是全名并且大小写敏感
     * @param tags
     * @param function 聚合器：如{@link Aggregators}，用于组合多个时间序列的数学函数（即如何合并一个组中的时间序列值）
     * @param rate     用于计算结果的每秒变化率
     * @return
     */
    public DataPoints[] query(long start, long end, String metric, Map<String, String> tags, Aggregator function,
            boolean rate) {
        Query query = tsdb.newQuery();
        query.setStartTime(start / 1000);
        query.setEndTime(end / 1000);
        query.setTimeSeries(metric, tags, function, rate);
        return query.run();
    }

}
